package cn.com.netedge.iot.frontend.datasave.job;

import cn.com.netedge.iot.common.constants.Globals;
import cn.com.netedge.iot.common.utils.RedisUtil;
import cn.com.netedge.iot.frontend.datasave.service.DataSaveService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public abstract class BaseSaveJob {
    private static final Logger logger = LoggerFactory.getLogger(BaseSaveJob.class);
    @Autowired
    protected DataSaveService dataSaveService;
    protected List<?> saveDataList;
    protected String cacheKey;


    public void chunkRun() {
        init();
        long start = System.currentTimeMillis();
        int count = 0;
        List<String> itemList = new ArrayList(Globals.CHUNK_RECORD_SIZE);

        try {
            List<String> objList = RedisUtil.range(cacheKey, 0, Globals.MAX_CACHE_SIZE);
            if (objList == null || objList.isEmpty()) {
                return;
            }
            RedisUtil.trim(cacheKey, objList.size() + 1, -1);
            for (String obj : objList) {
                itemList.add(obj);
                int batchSize = Globals.CHUNK_RECORD_SIZE;
                if (itemList.size() >= batchSize) {
                    count += itemList.size();
                    // 存库
                    batchSaveData(convertList(itemList));
                    itemList.clear();
                }
            }

            int size = itemList.size();
            if (size > 0) {
                count += size;
                // 存库
                batchSaveData(convertList(itemList));
            }
        } catch (Exception e) {
            logger.error("{} 消息大批量入库失败:", cacheKey, e);

            List<List<String>> partitionList = partition(itemList, Globals.CHUNK_SPLIT_SIZE);
            for (List<String> partition : partitionList) {
                try {
                    batchSaveData(convertList(partition));
                } catch (Exception ex) {
                    logger.error("{} 消息分区批量入库失败:", cacheKey, ex);
                    for (String item : partition) {
                        try {
                            batchSaveData(convertList(Collections.singletonList(item)));
                        } catch (Exception exx) {
                            logger.error("{} 消息单条入库失败:", cacheKey, exx);
                        }
                    }
                }
            }
        }

        if (count > 0) {
            long end = System.currentTimeMillis();
            logger.info("{} 消息入库:{} 条, 耗时:{} ms", cacheKey, count, (end - start));
            itemList.clear();
        }
    }

    private List<?> convertList(List<String> itemList) {
        List<?> dataList = new ArrayList<>(itemList.size());
        for (String item : itemList) {
            dataList.add(convert(item));
        }
        return dataList;
    }

    private List<List<String>> partition(List<String> list, int splitSize) {
        List<List<String>> result = new ArrayList<>(splitSize);
        int size = list.size();
        int chunkSize = size / splitSize;
        if (chunkSize <= 0) {
            splitSize = size;
            chunkSize = 1;
        }
        for (int i = 0; i < splitSize; i++) {
            int start = i * chunkSize;
            int end = start + chunkSize;
            if (i == splitSize - 1) {
                end = size;
            }
            if (start > end) {
                throw new IndexOutOfBoundsException("Index " + i + " is out of the list range <0," + (splitSize - 1) + ">");
            }
            List<String> sublist = new ArrayList<>(chunkSize);
            for (int j = start; j < end; j++) {
                sublist.add(list.get(j));
            }
            result.add(sublist);
        }
        return result;
    }

    protected abstract void init();

    protected abstract <T> T convert(String item);

    protected abstract void batchSaveData(List<?> saveDataList);
}
